package org.wangle.seckill.mq.config;


import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.wangle.seckill.mq.ack.ConfirmCallBackHandler;
import org.wangle.seckill.mq.ack.ReturnCallBackHandler;
import org.wangle.seckill.mq.component.death.SecKillDeathBinding;
import org.wangle.seckill.mq.component.death.SecKillDeathExchange;
import org.wangle.seckill.mq.component.death.SecKillDeathQueue;
import org.wangle.seckill.mq.component.delay.SecKillDelayBinding;
import org.wangle.seckill.mq.component.delay.SecKillDelayExchange;
import org.wangle.seckill.mq.component.delay.SecKillDelayQueue;
import org.wangle.seckill.mq.component.seckill.SecKillBinding;
import org.wangle.seckill.mq.component.seckill.SecKillExchange;
import org.wangle.seckill.mq.component.seckill.SecKillQueue;
import org.wangle.seckill.mq.customer.SecKillConsumer;
import org.wangle.seckill.mq.customer.SecKillDeathConsumer;

import javax.annotation.Resource;

@Configuration
@Import({
        SecKillBinding.class,
        SecKillExchange.class,
        SecKillQueue.class,
        SecKillDelayBinding.class,
        SecKillDelayExchange.class,
        SecKillDelayQueue.class,
        SecKillDeathBinding.class,
        SecKillDeathExchange.class,
        SecKillDeathQueue.class,
})
public class RabbitmqConfiguration {

    @Resource
    private CachingConnectionFactory connectionFactory;

    @Resource
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    @Bean(name = "simpleContainerFactory")
    public SimpleRabbitListenerContainerFactory simpleContainerFactory(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //容器工厂所用的实例
        factory.setConnectionFactory(connectionFactory);
        //设置消息的传输的格式，在这里采用Json的格式进行传输
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setPrefetchCount(100);
        //设置并发消费者最大的
        factory.setMaxConcurrentConsumers(10);
        //设置并发消费实例中每个实例拉取的消费数量为1
        factory.setReceiveTimeout(10000L);
        //设置消息的确认消费模式  NONE(无需确认机制) AUTO(自动确认机制) MANUAL(手动确认机制)
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

    @Bean(name = "simpleContainerManual")
    public SimpleMessageListenerContainer simpleContainerManual(){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        //容器工厂所用的实例
        container.setConnectionFactory(connectionFactory);
        //设置消息的传输的格式，在这里采用Json的格式进行传输
        //container.setMessageConverter(new Jackson2JsonMessageConverter());
        //设置并发消费者最大的
        container.setMaxConcurrentConsumers(1);
        //设置并发消费实例中每个实例拉取的消费数量为1
        container.setPrefetchCount(1);
        //设置消息的确认消费模式  NONE(无需确认机制) AUTO(自动确认机制) MANUAL(手动确认机制)
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return container;
    }



    @Bean
    public RabbitTemplate rabbitTemplate(){
        //设置消息发送哼过触发
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        //设置消息送达不成功触发
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        //rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        //消息发送成功时触发
        rabbitTemplate.setConfirmCallback(new ConfirmCallBackHandler());
        //消息发送失败时处罚
        rabbitTemplate.setReturnCallback(new ReturnCallBackHandler());

        return rabbitTemplate;
    }


    @Bean
    public RabbitAdmin rabbitAdmin(){
       return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public SecKillConsumer secKillConsumer1(){
        return new SecKillConsumer("consumer1");
    }


    @Bean
    public SecKillConsumer secKillConsumer2(){
        return new SecKillConsumer("consumer2");
    }



    @Bean
    public SecKillDeathConsumer secKillDeathConsumer1(){
        return new SecKillDeathConsumer("consumerDeath1");
    }


    @Bean
    public SecKillDeathConsumer secKillDeathConsumer2(){
        return new SecKillDeathConsumer("consumerDeath2");
    }

}
